package cn.dazhou.hellomqtt

import com.hivemq.client.mqtt.MqttGlobalPublishFilter
import com.hivemq.client.mqtt.datatypes.MqttQos
import com.hivemq.client.mqtt.mqtt3.Mqtt3AsyncClient
import com.hivemq.client.mqtt.mqtt3.Mqtt3Client
import com.hivemq.client.mqtt.mqtt3.message.publish.Mqtt3Publish
import java.util.concurrent.TimeUnit

object Mqtt {

    private const val clientId = "demo9"
    private const val host = "192.168.1.188"
    private const val port = 1883
    private const val topic = "message/topic"

    private var client: Mqtt3AsyncClient? = null

    private fun createMqttClient(): Mqtt3AsyncClient =
        Mqtt3Client.builder()
            .identifier(clientId)
            .serverHost(host)
            .serverPort(port)

            // 认证设置
            /*.simpleAuth()
            .username("admin")
            .password("password".toByteArray())
            .applySimpleAuth()*/

            // 重连设置
            .automaticReconnect()
            .initialDelay(1, TimeUnit.SECONDS) // 断线1秒后开始自动重连，如果重连还失败，则下次会等时间会按指数增长，比如2秒、4秒、8秒，双倍增长等待时间，但是不会超过最大值，由maxDelay函数来指定最大值。
            .maxDelay(32, TimeUnit.SECONDS)    // 断线后最多32秒就会自动重连，第5次连会来到32的位置，前面4次已用掉31秒的等待时间了。
            .applyAutomaticReconnect()

            // 连接状态监听器设置
            .addConnectedListener {
                println("MQTT${it.clientConfig.serverHost}:${it.clientConfig.serverPort}连接成功")
            }
            .addDisconnectedListener {
                // 客户端断开连接，或者连接失败都会回调这里
                println("MQTT${it.clientConfig.serverHost}:${it.clientConfig.serverPort}连接断开：${it.cause.message}，连接状态：${it.clientConfig.state.name}")
                /*when (it.clientConfig.state) {
                    MqttClientState.CONNECTING -> println("手动连接失败")             // 即主动调用connect时没连接成功
                    MqttClientState.CONNECTING_RECONNECT -> println("自动重连失败")   // 即连接成功后异常断开自动重连时连接失败
                    MqttClientState.CONNECTED -> println("连接正常断开或异常断开")
                    else -> println("连接断开：${it.clientConfig.state.name}")
                }*/
            }
            .buildAsync()

            // 消息监听器设置
            .also {
                // 接收订阅的消息。publishes必须在subscribe之前调用以确保消息不会丢失，可以在connect之前调用它以便接收前一个会话的消息。
                it.publishes(MqttGlobalPublishFilter.ALL) { publish: Mqtt3Publish ->
                    println("收到${publish.topic}的消息：${String(publish.payloadAsBytes)}")
                }
            }

    fun connect() {
        disconnect()
        // 断开连接后的client没法再复用，复用的client重新再连接时会收不到离线时的消息。所以每次连接时创建一个新的client。
        val client = createMqttClient().also { this.client = it }
        client.connectWith()
            .cleanSession(false) // false为持久会话，这样离线再上线时还能收到离线时别人推送的消息。
            .keepAlive(60)  // 心跳时间间隔，单位为秒
            .send()
            .whenComplete { ack, e ->
                if (ack != null) {
                    // 连接成功之后订阅主题
                    println("手动连接成功：$ack")
                    client.subscribeWith().topicFilter(topic).qos(MqttQos.EXACTLY_ONCE).send()
                } else if (e != null) {
                    println("手动连接失败: ${e.message}")
                }
            }
    }

    fun disconnect() {
        client?.let {
            it.disconnect().thenAccept { println("手动断开了连接") }
            client = null
        }
    }

    fun subscribe(topic: String = this.topic, qos: MqttQos = MqttQos.EXACTLY_ONCE) {
        val client = this.client ?: return
        client.subscribeWith().topicFilter(topic).qos(MqttQos.EXACTLY_ONCE).send()
    }

    fun unsubscribe(topic: String = this.topic) {
        val client = this.client ?: return
        client.unsubscribeWith().topicFilter(topic).send()
    }

    fun publish(message: String, topic: String = this.topic,  qos: MqttQos = MqttQos.EXACTLY_ONCE, retain: Boolean = false) {
        val client = this.client ?: return
        client
            .publishWith()
            .topic(topic)
            .qos(qos)
            .retain(retain)
            .payload(message.toByteArray())
            .send()
    }

    fun clearRetainMessage(topic: String = this.topic) {
        val client = this.client ?: return
        // 发送一条空的retain消息即可清除retain消息
        client.publishWith().topic(topic).retain(true).send()
    }

}